package com.liwen.service.file;


import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import redis.clients.jedis.ShardedJedis;

/**
 * @author: li.rui
 * @time: 2018-04-18 16:45
 * @description:
 */
public abstract class Consumer<T> implements Runnable {
    private Logger log = LoggerFactory.getLogger(this.getClass());

    private Storage<T> storage;

    private ShardedJedis jedis = null;

    public Consumer(Storage<T> storage) {
        this.storage = storage;
    }

    public Consumer(Storage<T> storage, ShardedJedis jedis) {
        this.storage = storage;
        this.jedis = jedis;
    }

    /**
     * 消费数据
     *
     * @param product
     */
    public abstract void consume(T product);

    /**
     * 休眠时间
     *
     * @return
     */
    protected long sleep() {
        return 0;
    }

    @Override
    public void run() {
        try {
            beforeStart();
            T product = null;
            while ((product = storage.getProduct()) != null) {
                try {
                    beforeConsume(product);
                    consume(product);
                    afterConsume(product);
                } catch (Exception e) {
                    log.error("exception",e);
                    onException(product, e);
                }
                if (sleep() > 0) {
                    try {
                        Thread.sleep(sleep());
                    } catch (InterruptedException e) {
                        log.error("exception",e);
                    }
                }
            }
            afterCompleted();
        } catch (Exception e) {
            log.error("exception",e);
        } finally {
            if (null != jedis)
                jedis.close();
        }

    }

    /**
     * 开始执行任务之前
     */
    protected void beforeStart() {

    }

    /**
     * 任务执行结束之后
     */
    protected void afterCompleted() {

    }

    /**
     * 发生异常时的处理
     *
     * @param product
     * @param e
     */
    protected void onException(T product, Exception e) {

    }

    /**
     * consume执行前处理
     *
     * @param product
     */
    public void beforeConsume(T product) {

    }

    /**
     * consume执行后的处理
     *
     * @param product
     */
    public void afterConsume(T product) {

    }

}
